package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 19:03
 */
public class DWD_10_DwdTradeRefundPaySucApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_10_DwdTradeRefundPaySucApp().init("DwdTradeRefundPaySucApp", 3009, 1, "DwdTradeRefundPaySucApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 需要join 四张表
        // 1. 从ods_db中过滤出退款成功数据   (注意区别退单和退款)  refund_payment  更新  refund_status=0702 旧数据不能是null
        // 2. 字典表
        // 3. 从订单表中过滤退款成功的订单数据
        // 4. 从退单数据中数据读取退款成功数据
        
        
        // TODO 3. 从 Kafka 读取 topic_db 数据，封装为 Flink SQL 表
        tEnv.executeSql(
            "create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +
                "`data` map<string, string>, " +
                "`old` map<string, string>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + SQLUtil.getKafkaDDL("dwd_trade_refund_pay_suc", "ods_db"));
        
        // TODO 4. 建立 MySQL-LookUp 字典表
        tEnv.executeSql(SQLUtil.getBaseDicLookUpDDL());
        
        // TODO 5. 读取 退款 表数据，并筛选退款成功数据
        Table refundPayment = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                "data['total_amount'] total_amount, " +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'refund_payment' " +
                "and `type` = 'update' " +
                "and data['refund_status'] = '0705' " +
                "and `old`['refund_status'] is not null"
        );
        
        tEnv.createTemporaryView("refund_payment", refundPayment);
        
        // TODO 6. 读取订单表数据并过滤退款成功订单数据
        Table orderInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['province_id'] province_id, " +
                "`old` " +
                "from topic_db " +
                "where `table` = 'order_info' " +
                "and `type` = 'update' " +
                "and data['order_status']='1006' " +
                "and `old`['order_status'] is not null"
        );
        
        tEnv.createTemporaryView("order_info", orderInfo);
        
        // TODO 7. 读取退单表数据并过滤退款成功数据
        Table orderRefundInfo = tEnv.sqlQuery(
            "select " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['refund_num'] refund_num, " +
                "`old` " +
                "from topic_db " +
                "where `table` = 'order_refund_info' " +
                "and `type` = 'update' " +
                "and data['refund_status']='0705' " +
                "and `old`['refund_status'] is not null"
            //             order_refund_info 表的 refund_status 字段值均为 null
        );
        
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        
        // TODO 8. 关联四张表获得退款成功表
        Table resultTable = tEnv.sqlQuery(
            "select " +
                "rp.id, " +
                "oi.user_id, " +
                "rp.order_id, " +
                "rp.sku_id, " +
                "oi.province_id, " +
                "rp.payment_type, " +
                "dic.dic_name payment_type_name, " +
                "date_format(rp.callback_time,'yyyy-MM-dd') date_id, " +
                "rp.callback_time, " +
                "ri.refund_num, " +
                "rp.total_amount, " +
                "rp.ts, " +
                "current_row_timestamp() row_op_ts " +
                "from refund_payment rp  " +
                "left join  " +
                "order_info oi " +
                "on rp.order_id = oi.id " +
                "left join " +
                "order_refund_info ri " +
                "on rp.order_id = ri.order_id " +
                "and rp.sku_id = ri.sku_id " +
                "join  " +
                "base_dic for system_time as of rp.proc_time as dic " +
                "on rp.payment_type = dic.dic_code ");
        tEnv.createTemporaryView("result_table", resultTable);
        
        // TODO 9. 创建 Upsert-Kafka dwd_trade_refund_pay_suc 表
        tEnv.executeSql(
            "create table dwd_trade_refund_pay_suc( " +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "sku_id string, " +
                "province_id string, " +
                "payment_type_code string, " +
                "payment_type_name string, " +
                "date_id string, " +
                "callback_time string, " +
                "refund_num string, " +
                "refund_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL("dwd_trade_refund_pay_suc"));
        
        // TODO 10. 将关联结果写入 Upsert-Kafka 表
        tEnv.executeSql(
            "insert into dwd_trade_refund_pay_suc select * from result_table"
        );
    }
    
}
